package com.me.study.jdk8.async;

import com.me.common.pojo.Student;
import com.me.common.util.SleepUtil;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.security.SecureRandom;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static org.junit.Assert.*;

/**
 * todo JDK8 异步编程：<p></p>
 *  1，带 async的方法 都是异步执行，不带 async的方法都是同步阻塞
 *  2，then 代表前一个动作正常完成后
 *  3，apply 代表对前一阶段的结果应用新的函数，参数可以是 Function，BiFunction
 *     accept 表示消费对前一阶段的结果，参数可以是 Consumer, BiConsumer
 *  4，handle 表示完成之后的处理（正常或异常），并返回新结果，参数可以是 Function，BiFunction
 *     whenComplete 也表示完成之后的处理（正常或异常），但是不会影响链中传递的结果（修改也不行），参数可以是 Consumer, BiConsumer
 *
 * @author ME
 * @date 2021/3/24
 */
public class CompletableFutureTest {

    private static final Logger LOGGER = LoggerFactory.getLogger(CompletableFutureTest.class);

    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(5, 10, 5,
            TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000));

    /**
     * 任务编排：
     * 主线程查询任务对象，
     * 然后多线程异步并行查询任务的负责人、所属系统并分别赋值给任务实体对应的属性
     */
    @Test
    public void taskChain() {
        try {
            CompletableFuture<Student> mainFuture = CompletableFuture.supplyAsync(() -> {
                SleepUtil.seconds(1);
                Student student = new Student();
                student.setId(1);
                return student;
            });

            // 异步设置名称
            CompletableFuture<Student> nameFuture = mainFuture.thenApplyAsync(student -> {
                SleepUtil.seconds(1);
                student.setName("Tom");
                return student;
            });

            // 异步设置生日
            CompletableFuture<Student> birthdayFuture = mainFuture.thenApplyAsync(student -> {
                SleepUtil.seconds(1);
                student.setBirthday(LocalDate.of(1992, 2, 17));
                return student;
            });

            // 结合所有结果
            CompletableFuture.allOf(nameFuture, birthdayFuture).join();
            System.out.println(mainFuture.join());
        } catch (CompletionException e) {
            throw new RuntimeException(e.getCause());
        }
    }

    /**
     * todo CompletableFuture 会自动执行，不需要 join()。
     */
    @Test
    public void test() {
        System.out.println("");
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync 创建有返回值的异步任务");
            if (new SecureRandom().nextBoolean()) {
                return "jjj";
            }
            else {
                throw new RuntimeException("假装出错了");
            }
        }).exceptionally(e -> {
            System.out.println("exceptionally 异常才执行，否则跳过此环节：此时可以处理错误");
            return "fail";
        }).thenApplyAsync(s -> {
            // do something
            System.out.println("编排在 exceptionally 之后的 thenApplyAsync");
            return s.toUpperCase();
        });

        SleepUtil.seconds(1);
        assertEquals("thenApply 返回值是处理之后的结果", "JJJ", future.join());
    }

    /** 等待任务全部完成 */
    @Test
    public void allOf() {
        List<String> strList = Arrays.asList("1", "2", "3", "4", "5", "6", "7", "8", "9");
        // List<CompletableFuture<Map<String, Object>>> fList = new ArrayList<>();
        // strList.forEach(str -> {
        //     CompletableFuture<Map<String, Object>> f = CompletableFuture.supplyAsync(() -> {
        //         // 执行耗时较长的业务代码，这里模拟一下
        //         SleepUtil.seconds(new SecureRandom().nextInt(4));
        //         Map<String, Object> map = new HashMap<>();
        //         map.put(str, RandomStringUtils.random(3, true, false));
        //         return map;
        //     }, executor);
        //     fList.add(f);
        // });
        List<CompletableFuture<Map<String, Object>>> fList = strList.stream().map(str -> {
                    return CompletableFuture.supplyAsync(() -> {
                        // 执行耗时较长的业务代码，这里模拟一下
                        SleepUtil.seconds(new SecureRandom().nextInt(4));
                        Map<String, Object> map = new HashMap<>();
                        map.put(str, RandomStringUtils.random(3, true, false));
                        return map;
                    }, EXECUTOR);
                }).collect(Collectors.toList());

        Map<String, Object> result = new HashMap<>();
        CompletableFuture<Void> future = CompletableFuture.allOf(fList.toArray(new CompletableFuture[0]))
            .whenComplete((v, throwable) -> {
                if (throwable != null) {
                    throwable.printStackTrace();
                }
                // do something
                fList.forEach(tempFuture -> {
                    result.putAll(tempFuture.getNow(new HashMap<>()));
                });
            });

        future.join();
        System.out.println("allOf 的结果: " + result);
    }

    @Test
    public void thenApplyAsync() {
        // 对前一阶段的结果异步应用函数
        CompletableFuture<String> cf = CompletableFuture.completedFuture("success").thenApplyAsync(s -> {
            assertTrue(Thread.currentThread().isDaemon());
            SleepUtil.ms(500);
            return s.toUpperCase();
        });
        assertNull(cf.getNow(null));
        assertEquals("thenApply 返回值是处理之后的结果", "SUCCESS", cf.join());
    }

    @Test
    public void thenAcceptAsync() {
        // 异步处理（消费）前一阶段的结果
        StringBuilder result = new StringBuilder();
        CompletableFuture<Void> future = CompletableFuture.completedFuture("success").thenAcceptAsync(str -> {
            result.append(str.toUpperCase());
            SleepUtil.ms(500);
        });
        assertNull("thenAccept 处理之后没有返回值", future.join());
        assertEquals("thenAccept 没有返回值，可以通过外部对象接收", "SUCCESS", result.toString());
    }

    /**
     * 二元依赖：cf1、cf2 都执行成功后执行
     */
    @Test
    public void thenCombine() {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            return "step1 result";
        }, EXECUTOR);
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            return "step2 result";
        });
        cf1.thenCombine(cf2, (result1, result2) -> {
            System.out.println(result1 + " ++++++ " + result2);
            return "step3 result";
        }).thenAccept(System.out::println);
    }

    /**
     * 二元依赖：cf1、cf2 都执行成功后执行
     */
    @Test
    public void thenCombineErr() {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            return "step1 result";
        }, EXECUTOR);
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            return err();
        });
        CompletableFuture<Void> cf3 = cf1.thenCombine(cf2, (result1, result2) -> {
            fail("step1、step2 都成功才会执行 thenCombine，任意失败都会跳过");
            return "step3 result";
        }).thenAccept(s -> {
            fail("thenCombine 执行成功才会执行 thenAccept，任意失败都会跳过");
            System.out.println(s);
        });

        try {
            cf3.join();
        } catch (CompletionException e) {
            e.getCause().printStackTrace();
        }
    }

    @Test
    public void exceptionally() {
        // 整个任务链正常完成，则跳过 exceptionally() 以正常值返回；任务链执行异常时，则执行此 exceptionally 做补偿处理；
        CompletableFuture<String> future = CompletableFuture.completedFuture("success").exceptionally(s -> {
                    // 干了些啥。。。
                    return "failed";
                });
        assertEquals("正常完成，则啥都不干 以相同的值正常返回", "success", future.join());

        // 链式调用
        CompletableFuture<String> errFuture = CompletableFuture.completedFuture("success").thenApplyAsync(str -> {
            System.out.println("阶段二：" + str);
            return err();
        }).thenApplyAsync(str -> {
            str = str.toUpperCase();
            System.out.println("阶段三：" + str);
            return str;
        }).exceptionally(err -> {
            // 执行过程异常了，返回个啥玩意
            System.out.println("阶段四：" + err.getMessage());
            return "failed";
        });
        assertEquals("异常完成时，则执行此 exceptionally 做一些补偿处理", "failed", errFuture.join());

        // 非链式调用，需要接收上一阶段的返回
        CompletableFuture<String> errFuture2 = CompletableFuture.supplyAsync(() -> {
            return err();
        });
        errFuture2 = errFuture2.exceptionally(err -> {
            // 执行过程异常了，返回个啥玩意
            return "failed";
        });
        assertEquals("异常完成时，则执行此 exceptionally 做一些补偿处理", "failed", errFuture2.join());
    }

    /** 假装抛了个异常 */
    public String err() {
        if (1 == 1) {
            throw new RuntimeException("假装异常了");
        }
        return "success";
    }

    /**
     * handle 处理整个结果，包括正常和异常返回
     */
    @Test
    public void handle() {
        // handle 表示对完成后的结果应用新函数，并返回处理后的结果，参数可以是 Function，BiFunction
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            SleepUtil.seconds(1);
            return "success";
        });
        CompletableFuture<String> exceptionHandler = future1.handle((s, th) -> {
            if (Objects.nonNull(th)) {
                assertEquals("主动异常完成 CompletableFuture", "主动 异常完成 future1", th.getMessage());
            }

            return "new result";
        });

        // 使用指定的异常对象，异常完成 CompletableFuture
        future1.completeExceptionally(new RuntimeException("主动 异常完成 future1"));
        assertTrue("future1 是被异常完成的", future1.isCompletedExceptionally());

        try {
            future1.join();
            fail("应该抛异常，不抛是不对的");
        } catch(CompletionException ex) {
            assertEquals("主动 异常完成 future1", ex.getCause().getMessage());
        }

        assertEquals("handle() 处理结果（含异常）后，返回新结果", "new result", exceptionHandler.join());
    }

    @Test(expected = ExecutionException.class)
    public void get() throws ExecutionException, InterruptedException {
        CompletableFuture<String> errFuture = CompletableFuture.supplyAsync(() -> {
            return err();
        });
        String result = errFuture.get();

        fail("get() 抛出的是 编译时异常");
    }

    @Test(expected = CompletionException.class)
    public void join() {
        CompletableFuture<String> errFuture = CompletableFuture.supplyAsync(() -> {
            return err();
        });
        String result = errFuture.join();

        fail("join() 抛出的是 运行时异常 CompletionException");
    }

    @Test
    public void whenComplete2() {
        // whenComplete 表示消费完成后的结果，并返回接收到的参数，参数可以是 Consumer, BiConsumer
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            SleepUtil.ms(100);
            return "success";
        }).whenComplete((s, th) -> {
            SleepUtil.seconds(1);
            if (th == null) {
                System.out.println("正常执行完后对结果做处理: " + s.toUpperCase());
            }
        }).thenApplyAsync(s -> {
            return s + "被修改改了";
        });

        System.out.println(future1.join());
    }

    @Test
    public void whenComplete1() {
        // whenComplete 表示消费完成后的结果，并返回接收到的参数，参数可以是 Consumer, BiConsumer
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            SleepUtil.seconds(1);
            return "success";
        }).whenComplete((s, th) -> {
            if (th == null) {
                s = s.toUpperCase();
                System.out.println("正常执行完后对结果做处理: " + s.toUpperCase());
            }
        });
        assertEquals("whenComplete 会根据结果做额外操作，但是不影响原来的返回值。即使在方法体中修改也不行",
                "success", future1.join());
        System.out.println();

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            SleepUtil.seconds(1);
            if (1 == 1) {
                throw new RuntimeException("假装执行异常了。。。");
            }
            return "success";
        }).whenComplete((s, th) -> {
            if (th != null) {
                System.out.println("CompletableFuture 异常了，然后一顿操作");;
            }
        });
        // SleepUtil.seconds(2);
        try {
            System.out.println("future2 还没执行完: " + future2.isDone());
            System.out.println("future2 没执行完当然不是异常完成咯: " + future2.isCompletedExceptionally());
            future2.join();
            fail("应该抛异常，不抛是不对的");
        } catch(CompletionException ex) {
            System.out.println("主线程阻塞获取 CompletableFuture 结果。");
            ex.getCause().printStackTrace();
        }
    }

    @Test
    public void whenComplete() {
        // whenComplete 表示消费完成后的结果，并返回接收到的参数，参数可以是 Consumer, BiConsumer
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            SleepUtil.seconds(1);
            return "success";
        });
        CompletableFuture<String> exceptionHandler = future1.whenComplete((s, th) -> {
            if (th != null) {
                System.out.println("被 completeExceptionally 主动异常完成");
            }
        });

        // 使用指定的异常对象，异常完成 CompletableFuture
        future1.completeExceptionally(new RuntimeException("主动 异常完成 future1"));
        assertTrue("future1 应该是被异常完成的", future1.isCompletedExceptionally());

        try {
            future1.join();
            fail("future1 被主动异常完成，应该抛异常，不抛是不对的");
        } catch(CompletionException ex) {
            assertEquals("主动 异常完成 future1", ex.getCause().getMessage());
        }

        try {
            exceptionHandler.join();
            fail("应该抛异常，不抛是不对的");
        } catch(CompletionException ex) {
            assertEquals("whenComplete() 处理结果（含异常）后，返回接收到的参数", "主动 异常完成 future1", ex.getCause().getMessage());
        }
    }

    @Test
    public void supplyAsync() {
        CompletableFuture<String> future = CompletableFuture.completedFuture("success");
        SleepUtil.seconds(1);
        assertEquals("success", future.join());
    }

    @Test
    public void runAsync() {
        // 创建无返回值的异步任务
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            // 执行异步任务
            SleepUtil.ms(500);
        });

        assertFalse(future.isDone());
        SleepUtil.ms(1000);
        assertTrue(future.isDone());
    }

    @Test
    public void completeExceptionally() {
        // CompletableFuture 执行过程中，正在执行的阶段继续执行（后续抛弃），但是最终结果被 主动异常结束
        // completeExceptionally 会抛弃后续阶段，包括 handle、whenComplete
        CompletableFuture<String> delay = CompletableFuture.supplyAsync(() -> {
            return "message";
        }).thenApplyAsync(s -> {
            // 假装执行了 2s
            SleepUtil.seconds(2);
            return s.toUpperCase();
        }).handleAsync((s, th) -> {
            if (Objects.nonNull(th)) {
                System.out.println("异步任务发生异常:" + th.getMessage());
            }
            return "new result";
        });
        // SleepUtil.seconds(2);// 睡眠 2s，则执行 completeExceptionally时 CompletableFuture 已完成，不生效
        delay.completeExceptionally(new RuntimeException("CompletableFuture 异常完成。。"));

        try {
            delay.join();
            fail("应该抛异常，不抛是不对的");
        } catch(CompletionException e) {
            assertEquals("CompletableFuture 异常完成。。", e.getCause().getMessage());
        }
    }

    @Test
    public void completedFuture() {
        CompletableFuture<String> future = CompletableFuture.completedFuture("success");
        // 如果当前任务未执行完成 则返回 hhh (传入的参数)，否则返回结果
        assertEquals("success", future.getNow("hhh"));

        // 创建延时执行的 future
        CompletableFuture<String> delay = CompletableFuture.completedFuture("message")
                .thenApplyAsync(String::toUpperCase, new DelayedExecutor(1, TimeUnit.SECONDS));
        // 如果当前任务未执行完成 则返回 hhh (传入的参数)，否则返回结果
        assertEquals("hhh", delay.getNow("hhh"));
    }

    static final class DelayedExecutor extends ScheduledThreadPoolExecutor {
        final long delay;
        final TimeUnit unit;

        DelayedExecutor(long delay, TimeUnit unit) {
            super(1);
            this.delay = delay;
            this.unit = unit;
        }

        @Override
        public void execute(Runnable r) {
            schedule(r, delay, unit);
        }
    }

}
